package com.ai.smlcore.util;

import java.nio.charset.Charset;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import com.ai.frame.bean.OutputObject;
import com.ai.frame.logger.Logger;
import com.ai.frame.logger.LoggerFactory;
import com.ai.smlcore.util.Constants.SYS_CONFIG;

/**
 * 基于MQTT协议的IM消息在线、离线消息推送功能
 *
 * （默认发送消息，采用MQTT3.1.1协议）
 *
 * @author ZSL
 *
 */
public class MQTTUtil {

    private static final Logger logger = LoggerFactory
            .getApplicationLog(MQTTUtil.class);

    /**
     * 消息接收者-分隔符
     */
    public static final String receivers_separatorChar=",";


    /**
     * 发送消息
     * -----------------------------------------------------------------------
     * receivers示例: /业务名称常量/分组/唯一编号,/业务名称常量/分组/唯一编号
     * msgContent示例：一些个性化的消息类型、消息编号可以通过特殊字符追加到内容中
     * -----------------------------------------------------------------------
     * （TODO:后期根据实际压力情况，修改SOCKET模式更改为单例长连接，维持心跳，减少连接销毁SOCKET.提升性能）
     *
     * @param receivers 接收人
     * @param msgContent 消息内容
     * @author Van_Q 2015年10月26日
     */
    public void sendMsg(String receivers, String msgContent) {
        OutputObject object = new OutputObject();
        object.setReturnCode("0");
        try {
            // 接收人数组
            String[] receiverArr = receivers.split(receivers_separatorChar);
            // 消息接收级别(只有一次，确保消息只到达一次)
            int qos = 1;
            // MQTT消息服务器REMOTE_MQTT_IP_PORT
            String broker = "tcp://"+PropertiesUtil.getString(SYS_CONFIG.REMOTE_MQTT_IP_PORT);

            // SOCKET标识（如果ClientID已经与服务器连接，则持有同样ClientID的旧有连接必须由服务器关闭后，新连接才能建立）
            String clientId = "SYS_USER_"+System.nanoTime();
            MemoryPersistence persistence = new MemoryPersistence();
            try {
                MqttClient sysClient = new MqttClient(broker, clientId, persistence);
                MqttConnectOptions connOpts = new MqttConnectOptions();
                connOpts.setCleanSession(true);
                logger.debug("开始连接服务器... ",broker);
                sysClient.connect(connOpts);
                logger.debug("已成功连接服务器","");
                logger.debug("准备开始发布消息: ", msgContent);

                for (String topic : receiverArr) {
                    MqttMessage message = new MqttMessage(msgContent.getBytes(Charset.forName("UTF-8")));
                    message.setQos(qos);
                    message.setRetained(true);
                    sysClient.publish(topic, message);
                }
                logger.debug("消息发布完毕","");
                sysClient.disconnect();
                logger.debug("与服务器断开连接","");
            } catch (MqttException me) {
                logger.info("MQTT消息发送异常:",me.getMessage());
                object.setReturnCode("-9999");
                object.setReturnMessage("MQTT消息发送异常");
            }
        } catch (Exception e) {
            logger.info("系统异常！",e.getMessage());
            object.setReturnCode("-9999");
            object.setReturnMessage("系统异常！");
        }
    }

    /**
     * 测试主函数入口
     *
     * @param args
     */
    public static void main(String[] args){
        for(int i=0;i<1000;i++){
            MQTTUtil mqttTool=new MQTTUtil();
            String receivers="/mail/friends/1001,/mail/friends/1002,/mail/friends/1003";
            //消息类型^消息编号^消息内容
            String msgContent="未接来电通知^12001^来自河南.郑州.二七区的移动手机号13803821225的电话"+i;
            mqttTool.sendMsg(receivers, msgContent);
            try {
                Thread.sleep((long)3 * 100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
